﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;

using JetBlack.IO;
using JetBlack.Net.Sockets;

using JetBlack.TopicBus.Messages;

namespace JetBlack.TopicBus.Adapters
{
    /// <summary>
    /// A delegate called when data has been received.
    /// </summary>
    /// <param name="topic">The topic of the data.</param>
    /// <param name="data">The data itself.</param>
    /// <param name="isImage">When <c>true</c> the data is a full data set of the topic, otherwise it is a change.</param>
    public delegate void DataCallback(string topic, IDictionary<string, object> data, bool isImage);

    /// <summary>
    /// A delegate called when a subscription has been forwarded to this client by way of a notification request.
    /// </summary>
    /// <param name="clientId">The id of the client subscribing to the topic.</param>
    /// <param name="topic">The topic to which the client is subscribing.</param>
    /// <param name="isAdd">When <c>true</c> the client is adding a subscription, otherwiase the subscription is being removed.</param>
    public delegate void ForwardedSubscriptionCallback(int clientId, string topic, bool isAdd);

    /// <summary>
    /// A delegate called when the client closes.
    /// </summary>
    /// <param name="isAbnormal">If <c>true</c> the client has been closed abnormally.</param>
    public delegate void ClosedCallback(bool isAbnormal);

    /// <summary>
    /// This is the primary interface to the distributor. All functionality is built upon this class.
    /// </summary>
    /// <remarks>
    /// A class interacts with the distributor in the following ways:
    /// <list type="table">
    ///     <listheader>
    ///         <term>Interaction</term>
    ///         <description>Description</description>
    ///     </listheader>
    ///     <item>
    ///         <term>Connection</term>
    ///         <description>
    ///             <para>
    ///                 A client manages it&apos;s connection to a distributor with <see cref="Connect"/> and <see cref="Close"/>.
    ///                 When the client is closed the client is informed through the <see cref="OnClosed"/> event.
    ///             </para>
    ///         </description>
    ///     </item>
    ///     <item>
    ///         <term>Subscription</term>
    ///         <description>
    ///             <para>
    ///                 A client can manage subscriptions to data with <see cref="AddSubscription"/> and <see cref="RemoveSubscription"/>.
    ///                 Data is delivered through the <see cref="OnData"/> event.
    ///             </para>
    ///             
    ///             <para>
    ///                 When some client publishes data to a topic to which this client has subscribed the distributor forwards the data to this client. The data has a parameter
    ///                 which indicates whether or not the data was an &quot;image&quot;. An image is considered to be all the data associated with the topic. If the image flag is <c>false</c>
    ///                 the data is considered a &quot;delta&quot;, or a change to data that has already been sent to a client.
    ///             </para>
    ///         </description>
    ///     </item>
    ///     <item>
    ///         <term>Notification</term>
    ///         <description>
    ///             <para>
    ///                 A client can manage the receipt of notifications to subscriptions with <see cref="AddNotification"/> and <see cref="RemoveNotification"/>.
    ///                 Notifications are delivered through the <see cref="OnForwardedSubscription"/> event.
    ///             </para>
    ///             
    ///             <para>
    ///                 When this client adds a notification to a topic, if some client adds a subscription to the topic, the subscription is forwarded to this client. This is useful
    ///                 in a number of cases. If the client is acting as a publisher, it will be informed of the topics on which clients wish data. The client may simply wish to audit
    ///                 subscription requests made.
    ///             </para>
    ///             
    ///             <para>
    ///                 The topic on which this client is requesting notification is specified as a regular expression.
    ///             </para>
    ///             
    ///             <para>
    ///                 The message received when a notification is delivered contains the &quot;id&quot; of the client making the subscription. This allows this client to send data
    ///                 directly to a specific client. This typically happens when a client requests data on a topic to which other clients have already subscribed. This client needs
    ///                 to send an image to this client, but not to the other clients.
    ///             </para>
    ///         </description>
    ///     </item>
    ///     <item>
    ///         <term>Publication</term>
    ///         <description>
    ///             <para>
    ///                 A client can send data through the distributor to other clients with <see cref="Publish"/> and <see cref="Send"/>.
    ///             </para>
    ///             
    ///             <para>
    ///                 Data that is &quot;published&quot; is delivered to all subscribing clients, while only a specific client receives data that is &quot;sent&quot;. Sending data
    ///                 to a specific client requires the client&apos;s id, which can be obtained through the notification mechanism.
    ///             </para>
    ///         </description>
    ///     </item>
    /// </list>
    /// </remarks>
    public class Client
    {
        /// <summary>
        /// This event is raised when data has been received by the client.
        /// </summary>
        public event DataCallback OnData;

        /// <summary>
        /// When a client has requested notification that a client has made a data subscription this event is raised.
        /// </summary>
        public event ForwardedSubscriptionCallback OnForwardedSubscription;

        /// <summary>
        /// The event raise when a client has been closed.
        /// </summary>
        public event ClosedCallback OnClosed;

        private Socket socket;
        private bool isConnected;
        private BlockingReader reader;
        private BlockingWriter writer;
        private Thread thread;

        /// <summary>
        /// The constructor for a client;
        /// </summary>
        public Client()
        {
            isConnected = false;
            socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            thread = new Thread(Dispatch);
        }

        /// <summary>
        /// Connect to a distributor.
        /// </summary>
        /// <param name="host">The host name for the machine on which the distributor is running.</param>
        /// <param name="port">The port on which the distributor is listening for connections.</param>
        /// <returns>If the connection is successful, <c>true</c> is returned; otherwise <c>false</c>.</returns>
        /// <remarks>
        ///     <para>
        ///         A client must connect before it can interact with the distributor.
        ///     </para>
        /// </remarks>
        /// <example>
        ///     How to connect.
        ///     <code source="../ExampleSubscriber/Program.cs" lang="CSharp" />
        /// </example>
        public bool Connect(string host, int port)
        {
            socket.Connect(host, port);
            reader = new BlockingReader(socket);
            writer = new BlockingWriter(socket);
            thread.Name = string.Format("interactor/{0}", socket.RemoteEndPoint);
            isConnected = true;
            thread.Start();

            return true;
        }

        /// <summary>
        /// Close the connection to the distributor.
        /// </summary>
        /// <remarks>
        /// An explicit disconnection will result in an <see cref="OnClosed"/> event with the parameter indicating the disconnection was not abnormal.
        /// If the connection was closed through some other error the paremeter will indicate an abnormal disconnect.
        /// </remarks>
        public void Close()
        {
            IsConnected = false;
            socket.Disconnect(false);
            thread.Join();
        }

        /// <summary>
        /// Subscribe to data published to the distributor.
        /// </summary>
        /// <param name="topic">The data topic.</param>
        /// <remarks>
        /// When a subscription is made data will be delivered through the <see cref="OnData"/> event. This event will indicate if
        /// the data that has been delivered is an &quot;image&quot; i.e. a full set of data, or a &quot;delta&quot; i.e. a change to
        /// full set. Typically publishers will send a full set as the first message to a client, although this is not enforced, and
        /// it is possible to receive these messages out of order.
        /// </remarks>
        /// <example>
        ///     An example subscriber.
        ///     <code source="../ExampleSubscriber/Program.cs" lang="CSharp" />
        /// </example>
        public void AddSubscription(string topic)
        {
            Write(new SubscriptionRequest(topic, true));
        }

        /// <summary>
        /// Remove a data subscription.
        /// </summary>
        /// <param name="topic">The data topic.</param>
        /// <remarks>
        /// This method cancels a subscription. It is possible that data received, but not yet sent by the distrubtor will still
        /// be forwarded to the client.
        /// </remarks>
        /// <example>
        ///     An example subscriber with subscription removal.
        ///     <code source="../ExampleSubscriber/Program.cs" lang="CSharp" />
        /// </example>
        public void RemoveSubscription(string topic)
        {
            Write(new SubscriptionRequest(topic, false));
        }

        /// <summary>
        /// Publish data to all subscribed clients.
        /// </summary>
        /// <param name="topic">The topic.</param>
        /// <param name="isImage">If true the data is considered to be a full image of the data; otherwise it contains only changes.</param>
        /// <param name="data">The changed data.</param>
        /// <remarks>
        /// The distributor will forward the data published by this method to all subscribing clients, including the flag indicating if the
        /// data constitutes an &quot;image&quot; (i.e. a full set) of the data, or a &quot;delta&quot; (i.e. a change).
        /// </remarks>
        /// <example>
        ///     An example publisher.
        ///     <code source="../ExamplePublisher/Program.cs" lang="CSharp" />
        /// </example>
        public void Publish(string topic, bool isImage, IDictionary<string, object> data)
        {
            Write(new SubscriberMessage(topic, isImage, data));
        }

        /// <summary>
        /// Send data to a specific client.
        /// </summary>
        /// <param name="clientId">The id of the client.</param>
        /// <param name="topic">The topic.</param>
        /// <param name="isImage">If true the data is considered to be a full image of the data; otherwise it contains only changes.</param>
        /// <param name="data">The data.</param>
        /// <remarks>
        /// An image is typically sent when a forwarded subscription request is received by the notification system.
        /// </remarks>
        /// <example>
        ///     An example notifier that sends data.
        ///     <code source="../ExampleNotifier/Program.cs" lang="CSharp" />
        /// </example>
        public void Send(int clientId, string topic, bool isImage, IDictionary<string, object> data)
        {
            Write(new ClientMessage(clientId, topic, isImage, data));
        }

        /// <summary>
        /// Request forwarded subscriptions from the distributor.
        /// </summary>
        /// <param name="topicPattern">The topic pattern to monitor.</param>
        /// <remarks>
        ///     <para>
        ///         The distributor will forward to this client any subscription request which matches the topic pattern.
        ///     </para>
        ///     <para>
        ///         The pattern is specified as a reqular expression. A regular expression pattern match is relatively expensive.
        ///         More performance from the distributor will be achived if the number of notifications requested is small, which
        ///         one might achive by making the pattern fairly general. Any extra processing done on the client will take load
        ///         away from the distributor.
        ///     </para>
        ///     <para>
        ///         If a subscribing client terminates without removing subscriptions, this will be handled by the distributer in
        ///         order to maintain a valid state.
        ///     </para>
        /// </remarks>
        /// <example>
        ///     An example notifier.
        ///     <code source="../ExampleNotifier/Program.cs" lang="CSharp" />
        /// </example>
        virtual public void AddNotification(string topicPattern)
        {
            Write(new NotificationRequest(topicPattern, true));
        }

        /// <summary>
        /// Remove a notification request.
        /// </summary>
        /// <param name="topicPattern">The topic pattern being monitored.</param>
        /// <remarks>
        /// It is possible that notifications will be sent after this call if subscriptions were sent to the distributor,
        /// but not yet forwarded, before the removal request was made.
        /// </remarks>
        /// <example>
        ///     An example notifier.
        ///     <code source="../ExampleNotifier/Program.cs" lang="CSharp" />
        /// </example>
        virtual public void RemoveNotification(string topicPattern)
        {
            Write(new NotificationRequest(topicPattern, false));
        }

        /// <summary>
        /// The connection state of the client.
        /// </summary>
        public bool IsConnected
        {
            get { lock (this) return isConnected; }
            set { lock (this) isConnected = value; }
        }

        /// <summary>
        /// The local endpoint.
        /// </summary>
        public IPEndPoint LocalEndPoint
        {
            get { return socket.LocalEndPoint as IPEndPoint; }
        }

        /// <summary>
        /// The remote end point.
        /// </summary>
        public IPEndPoint RemoteEndPoint
        {
            get { return socket.RemoteEndPoint as IPEndPoint; }
        }

        private void Dispatch()
        {
            bool abnormalDisconnect = true;

            while (IsConnected)
            {
                try
                {
                    byte[] byteArray = reader.Read();
                    MemoryStream memoryStream = new MemoryStream(byteArray);
                    Message message = Message.Read(memoryStream);

                    switch (message.MessageType)
                    {
                        case MessageType.SubscriberMessage:
                            RaiseOnData(message as SubscriberMessage);
                            break;
                        case MessageType.ClientMessage:
                            RaiseOnData(message as ClientMessage);
                            break;
                        case MessageType.ForwardedSubscriptionRequest:
                            RaiseForwardedSubscriptionRequest(message as ForwardedSubscriptionRequest);
                            break;
                        default:
                            throw new ArgumentException("invalid message type");
                    }
                }
                catch (System.IO.EndOfStreamException)
                {
                    IsConnected = false;
                    abnormalDisconnect = false;
                    continue;
                }
                catch (IOException ex)
                {
                    if (ex.InnerException != null && ex.InnerException is SocketException)
                    {
                        SocketException socketException = (SocketException)ex.InnerException;
                        if (socketException.ErrorCode == 10060)
                            continue;
                    }

                    if (IsConnected)
                    {
                        IsConnected = false;
                        abnormalDisconnect = true;
                    }
                    else
                        abnormalDisconnect = false;

                    continue;
                }
                catch (SocketException ex)
                {
                    if (ex.ErrorCode == 10060)
                        continue;
                    IsConnected = false;
                }
                catch
                {
                    if (IsConnected)
                    {
                        IsConnected = false;
                        abnormalDisconnect = true;
                    }
                    else
                        abnormalDisconnect = false;

                    continue;
                }
            }

            if (OnClosed != null)
                OnClosed(abnormalDisconnect);
        }

        private void Write(Message message)
        {
            lock (writer)
            {
                try
                {
                    MemoryStream memoryStream = new MemoryStream();
                    message.Write(memoryStream);
                    byte[] byteArray = memoryStream.ToArray();
                    writer.Write(byteArray);
                }
                catch (SocketException)
                {
                    IsConnected = false;
                }
            }
        }

        private void RaiseOnData(SubscriberMessage message)
        {
            RaiseOnData(message.Topic, message.Data, false);
        }

        private void RaiseOnData(ClientMessage message)
        {
            RaiseOnData(message.Topic, message.Data, true);
        }

        private void RaiseOnData(string topic, IDictionary<string, object> data, bool isImage)
        {
            if (OnData != null)
                OnData(topic, data, isImage);
        }

        private void RaiseForwardedSubscriptionRequest(ForwardedSubscriptionRequest message)
        {
            if (OnForwardedSubscription != null)
                OnForwardedSubscription(message.ClientId, message.Topic, message.IsAdd);
        }
    }
}
